package com.poetic.market.analysis;

import com.poetic.market.analysis.domain.AdClickLog;
import com.poetic.market.analysis.domain.BlackListWarning;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * <pre>
 *  黑名单处类：
 *      是否是黑名单保存在 状态中
 *      状态没有redis中的自动过期，因此需要启动定时器来触发失效。
 *      黑名单凌晨自动清空，第二天重新统计。
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class FilterBlackListUser extends KeyedProcessFunction<Tuple2<Long, Long>, AdClickLog, AdClickLog> {
    private int maxCount;

    private OutputTag<BlackListWarning> blackListOutputTag;

    public FilterBlackListUser(OutputTag<BlackListWarning> blackListOutputTag, int maxCount) {
        this.maxCount = maxCount;
        this.blackListOutputTag = blackListOutputTag;
    }

    // 保存当前用户对当前广告的点击量
    private ValueState<Long> countState;
    // 标记当前（用户，广告）作为key是否第一次发送到黑名单
    private ValueState<Boolean> firstSent;
    // 保存定时器触发的时间戳，届时清空重置状态
    private ValueState<Long> resetTime;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Long> countStateDesc = new ValueStateDescriptor<>("countState", TypeInformation.of(new TypeHint<Long>() {
        }), 0L);
        countState = getRuntimeContext().getState(countStateDesc);
        ValueStateDescriptor<Boolean> firstSentDesc = new ValueStateDescriptor<>("firstSent", TypeInformation.of(new TypeHint<Boolean>() {
        }),Boolean.FALSE);
        firstSent = getRuntimeContext().getState(firstSentDesc);

        ValueStateDescriptor<Long> resetTimeDesc = new ValueStateDescriptor<>("resetTime",
                TypeInformation.of(new TypeHint<Long>() {
        }), 0L);
        resetTime = getRuntimeContext().getState(resetTimeDesc);
    }

    @Override
    public void processElement(AdClickLog value, Context ctx, Collector<AdClickLog> out) throws Exception {
        Long curCount = countState.value();
        // 如果是第一次处理，注册一个定时器，每天 00：00 触发清除
        if (curCount == 0) {
            Long ts = (ctx.timerService().currentProcessingTime() / (24 * 60 * 60 * 1000) + 1) *
                    (24 * 60 * 60 * 1000);
            resetTime.update(ts);
            ctx.timerService().registerEventTimeTimer(ts);
        }
        // 如果计数已经超过上限，则加入黑名单，用侧输出流输出报警信息
        if (curCount > maxCount) {
            if (!firstSent.value()) {
                firstSent.update(true);

                BlackListWarning blackList = BlackListWarning.builder()
                        .userId(value.getUserId())
                        .adId(value.getAdId())
                        .msg("Click over " + maxCount + " times today.")
                        .build();

                ctx.output(blackListOutputTag, blackList);
            }
            return;
        }
        // 点击计数加 1
        countState.update(curCount + 1);
        out.collect(value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<AdClickLog> out) throws Exception {
        firstSent.clear();
        countState.clear();
    }
}
